/**
 * Copyright (c) 2021 OceanBase
 * OceanBase CE is licensed under Mulan PubL v2.
 * You can use this software according to the terms and conditions of the Mulan PubL v2.
 * You may obtain a copy of Mulan PubL v2 at:
 *          http://license.coscl.org.cn/MulanPubL-2.0
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PubL v2 for more details.
 */

#ifndef __OB_RS_RESTORE_STAT_H__
#define __OB_RS_RESTORE_STAT_H__

#include "lib/container/ob_array.h"
#include "lib/container/ob_array_iterator.h"
#include "rootserver/restore/ob_restore_info.h"
#include "rootserver/restore/ob_restore_table_operator.h"
#include "share/partition_table/ob_partition_table_operator.h"
#include "share/schema/ob_multi_version_schema_service.h"
#include "share/partition_table/ob_partition_table_iterator.h"
#include "share/backup/ob_physical_restore_info.h"

namespace oceanbase {
namespace common {
class ObMySQLProxy;
}
namespace rootserver {
struct PhysicalRestorePartition {
  friend class ObPhysicalRestoreStat;

public:
  PhysicalRestorePartition()
      : schema_id_(common::OB_INVALID_ID),
        partition_id_(common::OB_INVALID_INDEX),
        partition_cnt_(0),
        schema_paxos_cnt_(0),
        begin_(0),
        end_(0),
        table_cnt_(0)
  {}

  TO_STRING_KV(
      KT_(schema_id), K_(partition_id), K_(partition_cnt), K_(schema_paxos_cnt), K_(begin), K_(end), K_(table_cnt));

  common::ObPartitionKey get_key() const
  {
    common::ObPartitionKey key(schema_id_, partition_id_, partition_cnt_);
    return key;
  }
  int64_t get_replica_count() const
  {
    return end_ - begin_;
  }

public:
  uint64_t schema_id_;
  int64_t partition_id_;
  int64_t partition_cnt_;  // partition_cnt in meta_table
  int64_t schema_paxos_cnt_;
  int64_t begin_;      // replica position begin. (%all_replica_ array index)
  int64_t end_;        // replica position end. (%all_replica_ array index)
  int64_t table_cnt_;  // table cnt in pg, standalone partition's table_cnt_ is always 1
};

class ObRestoreStat {
public:
  ObRestoreStat(RestoreJob& job_info, common::ObMySQLProxy* sql_proxy);
  virtual ~ObRestoreStat();
  int gather_stat();

public:
  RestoreJob& job_info_;
  common::ObArray<PartitionRestoreTask> partition_task_;

private:
  /* functions */
  int fill_partitions();
  /* variables */
  ObRestoreTableOperator restore_op_;
  common::ObMySQLProxy* sql_proxy_;
  DISALLOW_COPY_AND_ASSIGN(ObRestoreStat);
};

class ObPhysicalRestoreStat {
public:
public:
  ObPhysicalRestoreStat(share::schema::ObMultiVersionSchemaService& schema_service, common::ObMySQLProxy& sql_proxy,
      share::ObPartitionTableOperator& pt_operator, const share::ObPhysicalRestoreJob& job_info, bool sys_only,
      volatile bool& stop);
  virtual ~ObPhysicalRestoreStat();
  int gather_stat();

private:
  int check_stop();
  int fill_sys_partitions();
  int fill_user_partitions();
  template <typename SCHEMA>
  int add_partition(const uint64_t schema_id, const int64_t table_cnt, const SCHEMA& schema);
  int add_partition(const int64_t partition_cnt, const int64_t paxos_replica_cnt, const int64_t table_cnt,
      const share::ObPartitionInfo& info);

public:
  const share::ObPhysicalRestoreJob& job_info_;
  common::ObArray<PhysicalRestorePartition> partitions_;
  common::ObArray<share::ObPartitionReplica> replicas_;
  share::ObRestoreProgressInfo statistic_;
  bool only_sys_;

private:
  common::ObMySQLProxy& sql_proxy_;
  share::schema::ObSchemaGetterGuard guard_;
  share::schema::ObMultiVersionSchemaService& schema_service_;
  share::ObPartitionTableOperator& pt_operator_;
  volatile bool stop_;
  DISALLOW_COPY_AND_ASSIGN(ObPhysicalRestoreStat);
};

// Use partial Partition/Replica structure to store partitions' physical restore info.
// Physical restore data tasks were generated by ObPhysicalRestoreScheduler, and were executed by ObRebalanceTaskMgr.
template <typename SCHEMA>
int ObPhysicalRestoreStat::add_partition(const uint64_t schema_id, const int64_t table_cnt, const SCHEMA& schema)
{
  int ret = common::OB_SUCCESS;
  int64_t paxos_replica_num = 0;
  share::ObTablePartitionIterator iter;
  if (OB_FAIL(check_stop())) {
    RS_LOG(WARN, "restore scheduler stop", K(ret));
  } else if (OB_FAIL(iter.init(schema_id, guard_, pt_operator_))) {
    RS_LOG(WARN, "table partition iterator init failed", K(ret));
  } else if (OB_FAIL(schema.get_paxos_replica_num(guard_, paxos_replica_num))) {
    RS_LOG(WARN, "fail to get paxos replica num", K(ret), K(schema_id));
  } else {
    share::ObPartitionInfo info;
    while (OB_SUCC(ret) && OB_SUCC(iter.next(info))) {
      if (OB_FAIL(check_stop())) {
        RS_LOG(WARN, "restore scheduler stop", K(ret));
      } else if (OB_FAIL(add_partition(schema.get_partition_cnt(), paxos_replica_num, table_cnt, info))) {
        RS_LOG(WARN, "add partition failed", K(ret), K(info));
      }
      info.reuse();
    }
    if (common::OB_ITER_END == ret) {
      ret = common::OB_SUCCESS;
    } else {
      RS_LOG(WARN, "iterator table partition failed", K(ret), K(iter));
    }
  }
  return ret;
}

}  // namespace rootserver
}  // namespace oceanbase
#endif /* __OB_RS_RESTORE_STAT_H__ */
//// end of header file
